package edu.poly.cs9223.assginment1;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.UUID;

import org.apache.commons.codec.binary.Base64;
import org.codehaus.jackson.map.JsonMappingException;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.sns.AmazonSNSClient;
import com.amazonaws.services.sns.model.SubscribeRequest;
import com.amazonaws.services.sns.model.SubscribeResult;
import com.amazonaws.services.sns.model.UnsubscribeRequest;
import com.amazonaws.services.sqs.AmazonSQSClient;
import com.amazonaws.services.sqs.model.CreateQueueRequest;
import com.amazonaws.services.sqs.model.CreateQueueResult;
import com.amazonaws.services.sqs.model.DeleteMessageRequest;
import com.amazonaws.services.sqs.model.DeleteQueueRequest;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.amazonaws.services.sqs.model.ReceiveMessageResult;
import com.amazonaws.services.sqs.model.SetQueueAttributesRequest;

/**
 *  Use Amazon Simple Queue Service (SQS) as an endpoint to SNS. The SNS topic can be configured in order to act as a producer for an SQS queue.
 * @author Kevin Zhao
 *
 */
public class TopicListener implements Runnable {
	Logger logger = LoggerFactory.getLogger(TopicListener.class);
	static final long DUPLICATE_MESSAGE_EXPIRATION_DURATION = 5*60*1000;	// 5 minutes
	
	AmazonSQSClient sqsClient = null;
	String queueUrl = null;

	String topicArn = null;
	AmazonSNSClient snsClient = null;
	String subscriptionArn = null;
	
	Thread listenerThread = null;

	public TopicListener(String awsAccessKey, String awsSecretKey, String topicArn) throws Exception {
		if (null==awsAccessKey || awsAccessKey.length()==0) {
			throw new IllegalArgumentException("awsAccessKey cannot be null or empty");
		}
		if (null==awsSecretKey || awsSecretKey.length()==0) {
			throw new IllegalArgumentException("awsSecretKey cannot be null or empty");
		}
		if (null==topicArn || topicArn.length()==0) {
			throw new IllegalArgumentException("topicArn cannot be null or empty");
		}

		this.topicArn = topicArn;
		
		// Connect to SQS
		this.sqsClient = new AmazonSQSClient(new BasicAWSCredentials(awsAccessKey, awsSecretKey));
		
		/*
		 * Build the temporary SQS queue into which topic messages will be delivered
		 */
		
		// Build the CreateQueueRequest
		String queueName = "tempQueue-" + UUID.randomUUID().toString();

		// Create the queue
		try {
			CreateQueueResult createQueueResult = this.sqsClient.createQueue(new CreateQueueRequest(queueName));

			this.queueUrl = createQueueResult.getQueueUrl(); 
		}
		catch (Exception e) {
			throw new Exception("Unable to create temporary queue: "+queueName, e); 
		}

		/*
		 * Create the subscription to forward topic messages to the temp queue
		 */
		String queueArn = buildQueueArnFromUrl(this.queueUrl);
		
		boolean success = false;
		try {
			// Connect to SNS
			this.snsClient = new AmazonSNSClient(new BasicAWSCredentials(awsAccessKey, awsSecretKey));

			// Create the SubscribeRequest
			SubscribeResult subscribeResult = snsClient.subscribe(new SubscribeRequest(topicArn, "sqs", queueArn));
			this.subscriptionArn = subscribeResult.getSubscriptionArn();
			success = true;
		}
		finally {
			if (!success) {
				// Delete the temporary queue
				this.sqsClient.deleteQueue(new DeleteQueueRequest(this.queueUrl));
			}
		}
		
		/*
		 * Change the security policy on the temp queue to allow messages from the topic
		 */
		String policyString =
			"{\n" +
			"   \"Version\":\"2008-10-17\",\n" +
			"   \"Id\":\"" + queueArn + "/policyId\",\n" +
			"   \"Statement\":[\n" +
			"      {\n" +
			"         \"Sid\":\"" + queueArn + "/statementId\",\n" +
			"         \"Effect\":\"Allow\",\n" +
			"         \"Principal\":{\"AWS\":\"*\"},\n" +
			"         \"Action\":\"SQS:SendMessage\",\n" +
			"         \"Resource\":\"" + queueArn + "\",\n" +
			"         \"Condition\":{\n" +
			"            \"StringEquals\":{\"aws:SourceArn\":\"" + this.topicArn + "\"}\n" +
			"         }\n" +
			"      }\n" +
			"   ]\n" +
			"}\n";

		TreeMap<String, String> propertyMap = new TreeMap<String,String>(); 
		propertyMap.put("Policy", policyString); 

		SetQueueAttributesRequest setQueueAttributesRequest = new SetQueueAttributesRequest(); 
		setQueueAttributesRequest.setQueueUrl(this.queueUrl); 
		setQueueAttributesRequest.setAttributes(propertyMap); 
		this.sqsClient.setQueueAttributes(setQueueAttributesRequest); 
		
		/*
		 * Create the listener thread
		 */
		this.listenerThread = new Thread(this);
		this.listenerThread.setName("TopicListener:"+topicArn);
		this.listenerThread.start();
	}

	public void close() throws Exception {
		// Stop the listener thread
		if (this.listenerThread != null) {
			this.listenerThread.interrupt();

			try {
				this.listenerThread.join();
			}
			catch (InterruptedException e) {
				// catch and ignore
			}

			this.listenerThread = null;
		}
		
		// Unsubscribe from the topic
		try {
			this.snsClient.unsubscribe(new UnsubscribeRequest(this.subscriptionArn));
		}
		catch (Exception e1) {
			// just eat the exception and continue shutting down 
		}
		
		// Delete the temporary queue
		try {
			this.sqsClient.deleteQueue(new DeleteQueueRequest(this.queueUrl));
		}
		catch (Exception e2) {
			// just eat the exception and continue shutting down 
		}
	}

	/*
	 * Run as a separate thread to watch for messages to arrive in the temp queue.
	 */
	public void run() {
		final long delays[] = {2000, 4000, 8000, 16000, 32000};
		
		try {
			int pollsWithoutResults = 0;
			for (;;) {
				// Wait to receive one or more message(s) from the temp queue 
				ReceiveMessageResult receiveMessageResult = null;
				try {
					receiveMessageResult = this.sqsClient.receiveMessage(new ReceiveMessageRequest(this.queueUrl));
				}
				catch (AmazonServiceException ase) {
					// TODO: Should probably log something here
					
					// Pause 5 seconds so we don't get into a hard loop
					Thread.sleep(5000);
				}
				catch (AmazonClientException ace) {
					// TODO: Should probably log something here
					
					// Pause 5 seconds so we don't get into a hard loop
					Thread.sleep(5000);
				}
				
				// Process any messages
				boolean processedMessage = false;
				if (null != receiveMessageResult) {
					List<Message> messages = receiveMessageResult.getMessages();
					if (null != messages) {
						for (Message message : messages) {
							processMessage(message);
							processedMessage = true;

							// Now that it has been processed, we can delete the message from the queue 
							try {
								this.sqsClient.deleteMessage(new DeleteMessageRequest(this.queueUrl, message.getReceiptHandle()));
							}
							catch (AmazonServiceException ase) {
								// TODO: Should probably log something here
							}
							catch (AmazonClientException ace) {
								// TODO: Should probably log something here
							}
						}
					}
				}			
				
				if (processedMessage) {
					// We just got one or more messages, so poll again immediately
					pollsWithoutResults = 0;
				}
				else if (pollsWithoutResults < delays.length) {
					Thread.sleep(delays[pollsWithoutResults]);
					
					++pollsWithoutResults;
				}
				else {
					Thread.sleep(60*1000);	// 1 minute
				}
			}
		}
		catch (InterruptedException ie) {
			// Eat the exception and exit from run() 
		}
	}

	static Map<String, Long> recentMessageMap = new HashMap<String, Long>();
	
	/**
	 * Actions for receiving the SNS notifications from SQS
	 * @param message
	 */
	private void processMessage(Message message) {
		String envelopeStringBase64Encoded = message.getBody();
		
		// Decode the base64 string
		String envelopeString = new String(Base64.decodeBase64(envelopeStringBase64Encoded.getBytes()));
		
		// Decode the envelope into a Map<String, Object)
		Map<String, String> envelopeMap = null;
		try {
			envelopeMap = new ObjectMapper().readValue(envelopeString, Map.class);
		}
		catch (JsonMappingException jme) {
			// TODO: Should probably log something here
		}
		catch (IOException ioe) {
			// TODO: Should probably log something here
		}
		
		String messageId = envelopeMap.get("MessageId");
		logger.debug("MessageId:{}",messageId);
		
		// Get the current time, since it will be used in several places below
		long currentTime = new Date().getTime();
		
		// Check for duplicate messages
		boolean processMessage = true;
		if (null != messageId) {
			if (recentMessageMap.containsKey(messageId)) {
				logger.debug("Duplicate message ignored");
				processMessage = false;
			}
			recentMessageMap.put(messageId, currentTime);
		}
		
		if (processMessage) {
			// Decode the message contents into a Map<String, Object)
			Map<String, Object> messageMap = null;
			try {
				messageMap = new ObjectMapper().readValue(envelopeMap.get("Message"), Map.class);
			}
			catch (JsonMappingException jme) {
				// TODO: Should probably log something here
			}
			catch (IOException ioe) {
				// TODO: Should probably log something here
			}
	
			// Forward to the topic listener
//			Main.onSNSTopicMessage(messageMap);
		}
		
		logger.debug("Before map size={}",new Integer(recentMessageMap.size()).toString());
		Long expirationTime = currentTime - DUPLICATE_MESSAGE_EXPIRATION_DURATION;
		for (Iterator<Long> iter=recentMessageMap.values().iterator(); iter.hasNext(); ) {
			Long entryTime = iter.next();
			
			if (entryTime < expirationTime) {
				logger.debug("Expiring entry from map");
				iter.remove();
			}
		}
		logger.debug("After map size={}",new Integer(recentMessageMap.size()).toString());
	}
	
	private String buildQueueArnFromUrl(String queueUrl) {
		String urlParts[] = queueUrl.split("/");
		
		String region = null;
		if (urlParts[2].startsWith("queue")) {
			region = "us-east-1";
		}
		else {
			String hostParts[] = urlParts[2].split("//.");
			region = hostParts[0];
		}
		String arn = "arn:aws:sqs:" + region + ":" + urlParts[3] + ':' + urlParts[4];
		
		return arn;
	}
}
